package simpledb.storage;

import simpledb.common.Database;
import simpledb.common.DbException;
import simpledb.transaction.TransactionId;
import simpledb.util.IteratorWrapper;

import java.io.*;
import java.util.Iterator;
import java.util.NoSuchElementException;

/**
 *
 */
public class HeapPage implements Page {

    final HeapPageId      pid;                             // 页面ID，唯一标识这个页面
    final TupleDesc       td;                              // 描述这个页面中元组的结构（字段类型）
    final byte            header[];                        // 位图，每个 bit 表示一个 slot 是否被使用
    final Tuple           tuples[];                        // 存储实际的元组对象数组
    final int             numSlots;                        // 当前页面中可以存放的最大元组数量

    byte[]                oldData;                         // 用于恢复旧状态的原始字节数据
    private final Byte    oldDataLock = new Byte((byte) 0); // 同步锁对象

    private TransactionId tid;                             // 最后修改此页的事务ID

    public HeapPage(HeapPageId id, byte[] data) throws IOException {
        this.pid = id;
        this.td = Database.getCatalog().getTupleDesc(id.getTableId());
        this.numSlots = getNumTuples();
        DataInputStream dis = new DataInputStream(new ByteArrayInputStream(data));

        // 分配并读取页头（header）
        header = new byte[getHeaderSize()];
        for (int i = 0; i < header.length; i++)
            header[i] = dis.readByte();

        tuples = new Tuple[numSlots];
        try {
            // 分配并读取页面中的元组
            for (int i = 0; i < tuples.length; i++)
                tuples[i] = readNextTuple(dis, i);
        } catch (NoSuchElementException e) {
            e.printStackTrace();
        }
        dis.close();

        setBeforeImage(); // 设置旧状态，用于恢复
    }

    private int getNumTuples() {
        return (BufferPool.getPageSize() * 8) / (td.getSize() * 8 + 1);
    }

    private int getHeaderSize() {
        return (int) Math.ceil(getNumTuples() / 8.0);
    }

    @Override
    public HeapPage getBeforeImage() {
        try {
            byte[] oldDataRef = null;
            synchronized (oldDataLock) {
                oldDataRef = oldData;
            }
            return new HeapPage(pid, oldDataRef);
        } catch (IOException e) {
            e.printStackTrace();
            System.exit(1);
        }
        return null;
    }

    @Override
    public void setBeforeImage() {
        synchronized (oldDataLock) {
            oldData = getPageData().clone();
        }
    }

    @Override
    public HeapPageId getId() {
        return this.pid;
    }

    private Tuple readNextTuple(DataInputStream dis, int slotId) throws NoSuchElementException {
        if (!isSlotUsed(slotId)) {
            // 如果 slot 未使用，则跳过对应的字节数
            for (int i = 0; i < td.getSize(); i++) {
                try {
                    dis.readByte();
                } catch (IOException e) {
                    throw new NoSuchElementException("error reading empty tuple");
                }
            }
            return null;
        }

        // 创建新的元组并设置 RecordId
        Tuple t = new Tuple(td);
        RecordId rid = new RecordId(pid, slotId);
        t.setRecordId(rid);

        // 读取每个字段
        try {
            for (int j = 0; j < td.numFields(); j++) {
                Field f = td.getFieldType(j).parse(dis);
                t.setField(j, f);
            }
        } catch (java.text.ParseException e) {
            e.printStackTrace();
            throw new NoSuchElementException("parsing error!");
        }

        return t;
    }

    /**
     * 生成表示该页内容的字节数组。
     * 用于将此页序列化到磁盘。
     * <p>
     * 不变条件是：可以将 getPageData 生成的字节数组传给 HeapPage 构造函数，
     * 并让它产生一个相同的 HeapPage 对象。
     *
     * @see #HeapPage
     * @return 表示该页内容的字节数组
     */
    @Override
    public byte[] getPageData() {
        int len = BufferPool.getPageSize(); // 获取缓冲池中每一页的标准大小（通常是 4096 字节）
        ByteArrayOutputStream baos = new ByteArrayOutputStream(len); // 创建字节输出流
        DataOutputStream dos = new DataOutputStream(baos); // 包装为数据输出流

        // 写入 header 部分（每个 bit 表示 slot 是否被使用）
        for (int i = 0; i < header.length; i++) {
            try {
                dos.writeByte(header[i]); // 将 header 的每个字节写入流
            } catch (IOException e) {
                // 这种异常理论上不会发生
                e.printStackTrace();
            }
        }

        // 写入 tuples 部分（逐个写入每个元组）
        for (int i = 0; i < tuples.length; i++) {

            // 如果 slot 没有使用，则填充 0 值作为占位符
            if (!isSlotUsed(i)) {
                for (int j = 0; j < td.getSize(); j++) {
                    try {
                        dos.writeByte(0);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
                continue;
            }

            // 如果 slot 被使用，则写入实际的元组字段值
            for (int j = 0; j < td.numFields(); j++) {
                Field f = tuples[i].getField(j); // 获取第 i 个元组的第 j 个字段
                try {
                    f.serialize(dos); // 序列化字段并写入流
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

        // 填充剩余空间以确保整个页面大小等于 BufferPool.getPageSize()
        int zerolen = BufferPool.getPageSize() - (header.length + td.getSize() * tuples.length);
        byte[] zeroes = new byte[zerolen];
        try {
            dos.write(zeroes, 0, zerolen); // 写入填充字节
        } catch (IOException e) {
            e.printStackTrace();
        }

        // 刷新流
        try {
            dos.flush();
        } catch (IOException e) {
            e.printStackTrace();
        }

        return baos.toByteArray(); // 返回完整的字节数组
    }

    /**
     * 静态方法，用于生成一个空的 HeapPage 的字节数组。
     * 用于向文件添加新的空页面。
     * 将此方法的结果传递给 HeapPage 构造函数将创建一个没有任何有效元组的 HeapPage。
     *
     * @return 返回的字节数组
     */
    public static byte[] createEmptyPageData() {
        int len = BufferPool.getPageSize();
        return new byte[len]; // 默认初始化为全 0
    }

    /**
     * 删除页面中指定的元组；相应的 header bit 应更新以反映它不再存储在任何页面上。
     * @throws DbException 如果这个元组不在这个页面上，或者 slot 已经为空
     * @param t 要删除的元组
     */
    public void deleteTuple(Tuple t) throws DbException {
        final RecordId recordId = t.getRecordId(); // 获取元组的记录 ID
        final HeapPageId pageId = (HeapPageId) recordId.getPageId(); // 获取对应的页面 ID
        final int tn = recordId.getTupleNumber(); // 获取元组在页面中的编号

        // 检查是否是本页面上的元组
        if (!pageId.equals(this.pid)) {
            throw new DbException("Page id not match");
        }

        // 检查 slot 是否已经被释放
        if (!isSlotUsed(tn)) {
            throw new DbException("Slot is not used");
        }

        // 标记 slot 为未使用，并将元组置为 null
        markSlotUsed(tn, false);
        this.tuples[tn] = null;
    }

    /**
     * 将指定的元组添加到页面中；该元组应更新以反映它现在存储在该页面上。
     * @throws DbException 如果页面已满（没有空闲 slot）或 TupleDesc 不匹配
     * @param t 要添加的元组
     */
    public void insertTuple(Tuple t) throws DbException {
        // 检查元组结构是否与本页面匹配
        if (!t.getTupleDesc().equals(this.td)) {
            throw new DbException("Tuple desc is not match");
        }

        // 寻找第一个可用的 slot 插入元组
        for (int i = 0; i < getNumTuples(); i++) {
            if (!isSlotUsed(i)) {
                markSlotUsed(i, true); // 标记 slot 为已使用
                t.setRecordId(new RecordId(this.pid, i)); // 设置记录 ID
                this.tuples[i] = t; // 插入元组
                return;
            }
        }

        // 找不到空 slot，抛出异常
        throw new DbException("The page is full");
    }

    /**
     * 将此页面标记为“脏”或“干净”，并记录是哪个事务执行了修改。
     */
    @Override
    public void markDirty(boolean dirty, TransactionId tid) {
        this.tid = dirty ? tid : null;
    }

    /**
     * 返回最后一次修改此页面的事务ID，如果页面不是脏页，则返回 null。
     */
    @Override
    public TransactionId isDirty() {
        return this.tid;
    }

    /**
     * 返回此页面中空闲 slot 的数量。
     */
    public int getNumEmptySlots() {
        int emptyNum = 0;
        for (int i = 0; i < getNumTuples(); i++) {
            if (!isSlotUsed(i)) {
                emptyNum++;
            }
        }
        return emptyNum;
    }

    /**
     * 如果指定位置的 slot 被使用，则返回 true。
     */
    public boolean isSlotUsed(int i) {
        int byteIndex = i / 8;
        int posIndex = i % 8;
        byte target = this.header[byteIndex];
        return (byte) (target << (7 - posIndex)) < 0;
    }

    /**
     * 抽象方法，用于填充或清空页面上的某个 slot。
     */
    private void markSlotUsed(int i, boolean value) {
        int byteIndex = i / 8;
        int posIndex = i % 8;
        byte v = (byte) (1 << posIndex);
        byte headByte = this.header[byteIndex];
        this.header[byteIndex] = value ? (byte) (headByte | v) : (byte) (headByte & ~v);
    }

    /**
     * 返回一个遍历该页面上所有元组的迭代器。
     * （注意：这个迭代器不会返回空 slot 中的内容！）
     */
    public Iterator<Tuple> iterator() {
        return new IteratorWrapper<>(this.tuples);
    }
}
